﻿using System.Collections.Concurrent;
using System.IO.Compression;

namespace VictoriaMetrics.Client
{
    public class VictoriaMetricsQueue
    {
        #region MyRegion

        private readonly VictoriaMetricsClient _client;
        private readonly int _queueLimit;
        private readonly int _batchPostingLimit;
        private readonly object _lock = new();

        private readonly ConcurrentQueue<QueueEntity> _queue = new();
        private readonly Timer _timer;
        private readonly SemaphoreSlim _slim = new(1);
        private readonly Dictionary<string, List<QueueEntity>> _dict = [];

        public bool Gzip { get; set; } = true;

        public CompressionLevel CompressionLevel { get; set; } = CompressionLevel.Optimal;

        public int SendBuffer { get; set; } = 100;

        #endregion

        public VictoriaMetricsQueue(VictoriaMetricsClient client, int queueLimit = 0, int periodSecond = 2, int batchPostingLimit = 1000)
        {
            _client = client;
            _queueLimit = queueLimit;
            _batchPostingLimit = batchPostingLimit;
            _timer = new Timer(_ => OnTick(), null, 0, periodSecond * 1000);
        }

        public event Action<QueueEntity> OnOutOfLimit;

        public event Action<List<QueueEntity>> OnError;

        public event Func<List<QueueEntity>, Task> OnErrorAsync;

        public bool Add(object model, string metricName = null)
        {
            if (model is MetricsEntity mc)
            {
                metricName = mc.metric["__name__"];
            }
            else if (model is MetricsPointData point)
            {
                metricName = point.MetricName;
            }
            else
            {
                if (string.IsNullOrEmpty(metricName))
                {
                    throw new Exception("metricName is null or empty.");
                }
            }

            var entity = new QueueEntity
            {
                MetricName = metricName,
                Data = model
            };

            if (_queueLimit > 0)
            {
                lock (_lock)
                {
                    if (_queue.Count > _queueLimit)
                    {
                        if (OnOutOfLimit != null)
                        {
                            try
                            {
                                OnOutOfLimit.Invoke(entity);
                            }
                            catch { }
                        }
                        return false;
                    }
                }
            }
            _queue.Enqueue(entity);
            return true;
        }

        private void AddToDict(QueueEntity entity)
        {
            string id;
            if (entity.Data is MetricsEntity mc)
            {
#if NET462
                id = string.Join("-", mc.metric.Values);
#else
                id = string.Join('-', mc.metric.Values);
#endif
            }
            else if (entity.Data is MetricsPointData point)
            {
#if NET462
                id = string.Join("-", point.Metric.Values) + "-p";
#else
                id = string.Join('-', point.Metric.Values) + "-p";
#endif
            }
            else
            {
                id = $"{entity.MetricName}-{entity.Data.GetType().FullName}";
            }

            var ok = _dict.TryGetValue(id, out var val);
            if (!ok)
            {
                _dict.Add(id, [entity]);
            }
            else
            {
                val.Add(entity);
            }
        }


        private async void OnTick()
        {
            var ok = await _slim.WaitAsync(0);
            if (!ok)
            {
                return;
            }

            var yes = false;
            try
            {
                for (int i = 0; i < _batchPostingLimit; i++)
                {
                    ok = _queue.TryDequeue(out var val);
                    if (!ok)
                    {
                        break;
                    }
                    AddToDict(val);
                    yes = true;
                }

                if (!yes)
                {
                    return;
                }

                foreach (var item in _dict)
                {
                    if (item.Value.Count > 0)
                    {
                        var first = item.Value[0];
                        if (first.Data is MetricsEntity)
                        {
                            await _client.ImportMetricsAsync(item.Value.Select(s => (MetricsEntity)s.Data), Gzip, CompressionLevel, SendBuffer);
                        }
                        else if (first.Data is MetricsPointData)
                        {
                            await _client.ImportPointAsync(item.Value.Select(s => (MetricsPointData)s.Data), Gzip, CompressionLevel, SendBuffer);
                        }
                        else
                        {
                            await _client.ImportEntityAsync(first.MetricName, item.Value.Select(s => s.Data), Gzip, CompressionLevel, SendBuffer);
                        }
                        item.Value.Clear();
                    }
                }
            }
            catch
            {
                if (!yes)
                {
                    return;
                }

                foreach (var item in _dict)
                {
                    if (item.Value.Count > 0)
                    {
                        if (OnError != null)
                        {
                            try
                            {
                                OnError.Invoke(item.Value);
                            }
                            catch { }

                        }
                        else if (OnErrorAsync != null)
                        {
                            try
                            {
                                await OnErrorAsync(item.Value);
                            }
                            catch { }
                        }
                        item.Value.Clear();
                    }
                }
            }
            finally
            {
                _slim.Release();
            }
        }

        public async Task StopAsync()
        {
            try
            {
                _timer.Dispose();
                _slim.Wait();

                foreach (var item in _dict)
                {
                    if (item.Value.Count > 0)
                    {
                        item.Value.Clear();
                    }
                }

                var yes = false;
                while (_queue.TryDequeue(out var val))
                {
                    AddToDict(val);
                    yes = true;
                }

                if (yes)
                {
                    foreach (var item in _dict)
                    {
                        if (item.Value.Count > 0)
                        {
                            var first = item.Value[0];
                            if (first.Data is MetricsEntity)
                            {
                                await _client.ImportMetricsAsync(item.Value.Select(s => (MetricsEntity)s.Data), Gzip, CompressionLevel, SendBuffer);
                            }
                            else if (first.Data is MetricsPointData)
                            {
                                await _client.ImportPointAsync(item.Value.Select(s => (MetricsPointData)s.Data), Gzip, CompressionLevel, SendBuffer);
                            }
                            else
                            {
                                await _client.ImportEntityAsync(first.MetricName, item.Value.Select(s => s.Data), Gzip, CompressionLevel, SendBuffer);
                            }
                            item.Value.Clear();
                        }
                    }
                }
            }
            catch { }
            finally
            {
                _slim.Release();
                _slim.Dispose();
            }
        }
    }
}
